package RMService;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.thrift.TException;
import TopologyService.Topology;

import KVService.KVClient;
import KVService.KVHandler;
import RMService.RMService.Iface;


public class RMHandler implements Iface {

	// Member variables
	private Topology topology;
	private int ttl;
	KVHandler kvhandler;
	
	/**
	 * Constructor for RMHandler object.
	 * 
	 * @param topology the topology object
	 * @param ttl the ttl for inserts and deletes
	 * @throws TException 
	 */
	public RMHandler(Topology topology, int ttl, KVHandler kvhandler) {
		this.topology = topology;
		this.ttl = ttl;
		this.kvhandler = kvhandler;
		
	}
	
	/**
	 * Function to lookup the owner process of a key.
	 * 
	 * @param key the key value to search for
	 * @return the id of the owner process (node)
	 * @throws TException
	 */
	@Override
	public int lookup(int key) throws TException {
		return kvhandler.lookup(key);
	}

	/**
	 * Function to insert a key,value pair.
	 * 
	 * @param key the key for the pair
	 * @param value the value for the pair
	 * @throws TException
	 */
	@Override
	public void insert(int key, String value) throws TException {
		// insert key (this node owns the key because a lookup of the key was done before calling insert)
		kvhandler.insert(key, value);
		
		// replicate the key
		// origin = -1 since we are initiating the broadcast
		replicate(key, value, ttl-1, -1);
	}
	
	/**
	 * Initialize replication management (get keys for this node's buckets from neighboring ndoes)
	 * This funtion is called by Node.java when the node has been connected to the topology
	 * @throws TException 
	 */
	public void init() throws TException{
		// use -1 origin as we are initiating message around ring
		Map<Integer, String> keyValues = new HashMap<Integer,String>();
		getKeys(kvhandler.getKeyBuckets(), keyValues, -1, ttl);
	}

	/**
	 * Function to delete the key,value pair.
	 * 
	 * @param key the key of the key,value pair
	 * @param ttl the value to prevent deleteKey messages from being sent continually
	 * @throws TException
	 */
	@Override
	public void deleteKey(int key) throws TException {
		// delete key (this node owns the key because a lookup of the key was done before calling delete)
		kvhandler.deleteKey(key);
		
		// delete replicas
		// origin = -1 since we are initiating the broadcast
		deleteReplicas(key, ttl - 1, -1);
	}

	@Override
	public void replicate(int key, String value, int ttl, int origin) throws TException {
		
		if(kvhandler.ownsKey(key) && origin != -1){
			kvhandler.insert(key, value);
			System.out.println("Process " + topology.myID() + " replicating key " + key +" in bucket " + Math.abs(key %4));
		}
		
		if(ttl > 0 && origin != topology.myID()){
			if(origin == -1){
				origin = topology.myID();
			}
			boolean sent = false;
		    while(!sent){
		    	String ip = topology.getNextIP();
		    	int port = topology.getNextPort();
		    	int routedNode = topology.getNextID();
				try {
					
					RMClient nextRM = new RMClient(ip, port);
					System.out.println("Process " + topology.myID() + " sending replica of key " + key + " to " + routedNode + " for bucket " + Math.abs(key %4) + "origin: " + origin);
					nextRM.replicate(key, value, ttl -1, origin);
					sent = true;
				} catch (TException e) {
					// fix topology and try again
					topology.failureDetected(routedNode);
				}
		    }

		}
		else{
			System.out.println("Process " + topology.myID() + " killing replica message key: " + key + " ttl: " + ttl + " owner: " + (origin == topology.myID()));
		}
	}


	@Override
	public String get(int key) throws TException {
		return kvhandler.get(key);
	}

	@Override
	public void deleteReplicas(int key, int ttl, int origin)
			throws TException {
		if(kvhandler.ownsKey(key) && origin != -1){
			kvhandler.deleteKey(key);
			System.out.println("Process " + topology.myID() + " deleting key in bucket " + Math.abs(key %4));
		}
		
		
		if(ttl > 0 && origin != topology.myID()){
			if(origin == -1){
				origin = topology.myID();
			}
			boolean sent = false;
		    while(!sent){
		    	String ip = topology.getNextIP();
		    	int port = topology.getNextPort();
		    	int routedNode = topology.getNextID();
				try {
					
					RMClient nextRM = new RMClient(ip, port);
					System.out.println("Process " + topology.myID() + " sending delete replica to " + routedNode + " for bucket " + Math.abs(key %4) +" origin: " + origin);
					nextRM.deleteReplicas(key, ttl -1, origin);
					sent = true;
				} catch (TException e) {
					// fix topology and try again
					topology.failureDetected(routedNode);
				}
		    }
		}
		else{
			System.out.println("Process " + topology.myID() + " killing replica delete message ttl: " + ttl + " owner: " + (origin == topology.myID()));
		}
		
	}

	@Override
	public void getKeys(List<Integer> buckets, Map<Integer, String> keyValues, int origin, int ttl) throws TException {
		
		if(origin == topology.myID()){
			// message has made it around the ring, initialize kv service with (key,values) returned from ring
			kvhandler.setKeyStore(keyValues);
			System.out.println("Process " + topology.myID() + " initialized key store with " + keyValues.size() + " keys");
			return;
		}
		
		
		if(origin == -1){
			// we are being called from the same node
			origin = topology.myID();
		}
		
		if(ttl > 0){
			
		
			// add our keys to keyValues and pass message around ring
			Map<Integer, String> kvstore = kvhandler.getKeyStore();
			Iterator it = kvstore.entrySet().iterator();
			while(it.hasNext()){
				Map.Entry<Integer, String> pair = (Map.Entry)it.next();
				int key = pair.getKey();
				int bucket = Math.abs(key % 4);
				if(buckets.contains(bucket)){
					keyValues.put(key, pair.getValue());
				}
			}
			
			// send updated key values along ring
			boolean sent = false;
		    while(!sent){
		    	String ip = topology.getNextIP();
		    	int port = topology.getNextPort();
		    	int routedNode = topology.getNextID();
				try {
					
					RMClient nextRM = new RMClient(ip, port);
					System.out.println("Process " + topology.myID() + " sending get keys to " + routedNode + " origin: " + origin);
					nextRM.getKeys(buckets, keyValues, origin, ttl-1);
					sent = true;
				} catch (TException e) {
					// fix topology and try again
					topology.failureDetected(routedNode);
				}
		    }
		
		}
		else{
			System.out.println("Process " + topology.myID() + " killing getKeys message ttl: " + ttl + " owner: " + (origin == topology.myID()));
		}
	}
}
